package com.yangluhan.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.yangluhan.util.DateFormatUtil;
import com.yangluhan.util.MyKafkaUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //TODO 2.接收topic_log数据
        String topic = "topic_log";
        String groupID =  "base_db_app";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupID));
        //TODO 3.过滤非json数据,输出到测流,并将每行转为json输出
        OutputTag<String> dirty = new OutputTag<String>("Dirty") {
        };
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> out) throws Exception {
                        try {
                            JSONObject jsonObject = JSON.parseObject(value);
                            out.collect(jsonObject);
                        } catch (Exception e) {
                            context.output(dirty, value);
                        }
                    }
                }
        );
        DataStream<String> dirtyDS = jsonObjDS.getSideOutput(dirty);
        dirtyDS.print("Dirty>>>>>>");
        //TODO 4.按照Mid分组
        KeyedStream<JSONObject, String> keyedStream = jsonObjDS.keyBy(json -> json.getJSONObject("common").getString("mid"));
        //TODO 5.使用状态编程做新老访客标记校验
        SingleOutputStreamOperator<JSONObject> jsonObjWhitNewFlagDs = keyedStream.map(
                new RichMapFunction<JSONObject, JSONObject>() {
                    private ValueState<String> lastVisitState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        lastVisitState = getRuntimeContext().getState(new ValueStateDescriptor<String>("last_visit", String.class));
                    }

                    @Override
                    public JSONObject map(JSONObject value) throws Exception {
                        String isNew = value.getJSONObject("common").getString("is_new");
                        Long ts = value.getLong("ts");
                        String curDate = DateFormatUtil.toDate(ts);
                        //获取状态中的日期
                        String lastDate = lastVisitState.value();
                        if ("1".equals(isNew)) {
                            if (lastDate == null) {
                                lastVisitState.update(curDate);
                            } else if (!lastDate.equals(curDate)) {
                                value.getJSONObject("common").put("is_new", 0);
                            }
                        } else if (lastDate == null) {
                            lastVisitState.update(DateFormatUtil.toDate(ts - 24 * 60 * 60 * 1000L));
                        }
                        return value;
                    }
                }
        );
        //TODO 6.使用侧输出流进行分流处理  页面日志放到主流  启动、曝光、动作、错误放到侧输出流
        OutputTag<String> startTag = new OutputTag<String>("start") {};
        OutputTag<String> actionsTag = new OutputTag<String>("actions") {};
        OutputTag<String> displaysTag = new OutputTag<String>("displays") {};
        OutputTag<String> errorTag = new OutputTag<String>("error") {};
        SingleOutputStreamOperator<String> pageDS = jsonObjWhitNewFlagDs.process(
                new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject value, ProcessFunction<JSONObject, String>.Context ctx, Collector<String> out) throws Exception {
                        //尝试获取error数据
                        String error = value.getString("err");
                        //判断是否有error数据
                        if (error != null) {
                            //将带有error的数据写入到errorTag流
                            ctx.output(errorTag, value.toJSONString());
                        }
                        //删除error数据
                        value.remove("err");

                        //尝试获取start数据
                        String start = value.getString("start");
                        if (start != null) {
                            ctx.output(startTag, value.toJSONString());
                        } else {
                            String common = value.getString("common");
                            String pageID = value.getJSONObject("page").getString("page_id");
                            Long ts = value.getLong("ts");

                            //尝试获取曝光日志
                            JSONArray displays = value.getJSONArray("displays");
                            if (displays != null && displays.size() > 0) {
                                for (int i = 0; i < displays.size(); i++) {
                                    JSONObject jsonObject = displays.getJSONObject(i);
                                    jsonObject.put("common", common);
                                    jsonObject.put("page_id", pageID);
                                    jsonObject.put("ts", ts);
                                    ctx.output(displaysTag, displays.toJSONString());
                                }
                            }
                            //尝试获取动作日志
                            JSONArray actions = value.getJSONArray("actions");
                            if (actions != null && actions.size() > 0) {
                                for (int i = 0; i < actions.size(); i++) {
                                    JSONObject jsonObject = actions.getJSONObject(i);
                                    jsonObject.put("common", common);
                                    jsonObject.put("page_id", pageID);
                                    ctx.output(actionsTag, actions.toJSONString());
                                }
                            }
                            value.remove("displays");
                            value.remove("actions");
                            out.collect(value.toJSONString());
                        }
                    }
                }
        );
        //TODO 7.提取各个侧输出流数据
        DataStream<String> startDS = pageDS.getSideOutput(startTag);
        DataStream<String> displayDS = pageDS.getSideOutput(displaysTag);
        DataStream<String> actionDS = pageDS.getSideOutput(actionsTag);
        DataStream<String> errorDS = pageDS.getSideOutput(errorTag);
        //TODO 8.将数据打印并写入对应的主题
        pageDS.print("Page>>>>>>>>>>");
        startDS.print("Start>>>>>>>>");
        displayDS.print("Display>>>>");
        actionDS.print("Action>>>>>>");
        errorDS.print("Error>>>>>>>>");

        String page_topic = "dwd_traffic_page_log";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String error_topic = "dwd_traffic_error_log";

        pageDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(page_topic));
        startDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(start_topic));
        displayDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(display_topic));
        actionDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(action_topic));
        errorDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(error_topic));
        //TODO 9.启动任务
        env.execute("BaseLogApp");
    }
}
